#include "aio_udp.h"
#include "../global/head.h"
#include "../xos/head.h"
#include "../buf/head.h"
#include "aio_callback_thread.h"
#include "../msg/head.h"
#include "aio_impl.h"
#include "aio_key.h"

/////////////////////////////////////////////////////////////////////////////////////////
// 
namespace xos
{
    aio_udp::aio_udp( aio_impl * pAioImpl ) : m_pAioImpl( pAioImpl )
    {
    }

    aio_udp::~aio_udp()
    {
    }

    ////////////////////////////////////////////////////////////////////////////////////
    // 

    // 
    // m_nData[0] : cmd type
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::close_handle( msg_impl * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->m_lpData[1];
        udp_impl * lpUdp = ( udp_impl * )lpKey->m_pHandle;
        socket_impl * lpSocket = &lpUdp->m_socket;
        int hSocket = lpSocket->m_hSocket;

        {
            struct kevent change[1] = { 0 };
            EV_SET( &change[0], hSocket, EVFILT_READ, EV_DELETE, 0, 0, this );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
            lpKey->finish_request( m_pAioImpl );
            lpKey->release();
            lpKey = 0;
        }

        // 
        // return.
        // 
        lpMsg->m_nData[0] = i_aio::AIO_UDP_CLOSE_HANDLE_RET;
        lpMsg->m_bData[0] = true;

        m_pAioImpl->m_pCallbackThread->notify( lpMsg );
        lpMsg = 0;

        return nRet;
    }

    // 
    // m_bData[0] : true or false for result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::user_init_server( msg_impl * lpMsg )
    {
        using namespace std;
        int nRet = 0;

        socket_impl * lpSocket = 0;
        udp_impl * lpUdp = 0;
        aio_key * lpKey = 0;
        char szIp[100] = { 0 };
        xos_u16 usPort = 0;
        int hSocket = -1;

        if( 0 == nRet )
        {
            nRet = g_pXosImpl->create( i_xos::XOS_OBJ_UDP, ( void ** )&lpUdp );
        }

        if( 0 == nRet )
        {
            lpSocket = &lpUdp->m_socket;
            nRet = lpUdp->m_socket.udp_init( lpMsg->m_szStr[1], lpMsg->m_nData[1] );
            if( 0 == nRet )
            {
                lpSocket->set_async( true );
            }
        }

        if( 0 == nRet )
        {
            lpKey = m_pAioImpl->m_key_pool.GetItem();
            lpKey->init();
            lpKey->m_eState = aio_key::KEY_STATE_RUNNING;
            lpKey->m_lpUserKey = lpMsg->m_lpData[0];
            lpKey->m_pHandle = lpUdp;
            hSocket = lpSocket->m_hSocket;
        }

        if( 0 == nRet )
        {
            lpKey->m_nKqueueFlags = EVFILT_READ | EVFILT_WRITE;
            struct kevent change[2] = { 0 };
            EV_SET( &change[0], hSocket, EVFILT_WRITE, EV_ADD | EV_ENABLE, 0, 0, lpKey );
            EV_SET( &change[1], hSocket, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, lpKey );
            kevent( m_pAioImpl->m_hKqueue, change, 2, NULL, 0, NULL );
        }

        if( 0 == nRet )
        {
            //lpSocket->get_socket_addr( szIp, &usPort );
            strcpy( lpMsg->m_szStr[1], szIp );
            lpMsg->m_nData[1] = usPort;
            lpMsg->m_lpData[1] = lpKey;
            lpMsg->m_bData[0] = true;
            lpUdp = 0;
            lpKey = 0;
        }
        else
        {
            lpMsg->m_bData[0] = false;
        }

        lpMsg->m_nData[0] = i_aio::AIO_UDP_INIT_RET;
        m_pAioImpl->m_pCallbackThread->notify( lpMsg );
        lpMsg = 0;

        if( lpUdp )
        {
            lpUdp->release();
            lpUdp = 0;
        }

        if( lpKey )
        {
            lpKey->m_pHandle = 0;
            lpKey->release();
            lpKey = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : peer port
    // m_szStr[1] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::user_send( msg_impl * lpMsg )
    {
        using namespace std;
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->m_lpData[1];

        lpMsg->m_nData[0] = i_aio::AIO_UDP_SEND_RET;
        lpMsg->m_bData[0] = false;
        lpKey->m_pend_send_list.push_back( lpMsg );

        if( aio_key::KEY_STATE_RUNNING != lpKey->m_eState )
        {
            lpKey->finish_request( m_pAioImpl );
        }
        else
        {
            udp_send( lpKey );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : peer port
    // m_szStr[1] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::user_recv( msg_impl * lpMsg )
    {
        using namespace std;
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->m_lpData[1];
        msg_impl * lpmsg = 0;

        lpMsg->m_nData[0] = i_aio::AIO_UDP_RECV_RET;
        lpMsg->m_bData[0] = false;
        buf_impl * lpBuf = lpMsg->m_lpBuf[0];

        if( lpKey->m_recv_list.size() > 0 )
        {
            lpmsg = lpKey->m_recv_list.front();
            lpKey->m_recv_list.pop_front();
        }

        if( lpmsg )
        {
            buf_impl *& lpbuf = lpmsg->m_lpBuf[0];

            lpMsg->m_bData[0] = lpmsg->m_bData[0];
            lpMsg->m_nData[0] = lpmsg->m_nData[0];
            lpMsg->m_nData[1] = lpmsg->m_nData[1];
            lpMsg->m_nData[2] = lpmsg->m_nData[2];

            strcpy( lpMsg->m_szStr[1], lpmsg->m_szStr[1] );
            lpMsg->m_lpData[1] = lpmsg->m_lpData[1];
            memcpy( lpBuf->m_szBuf, lpbuf->m_szBuf + lpbuf->m_nPos, lpbuf->m_nLen );
            lpBuf->m_nLen = lpbuf->m_nLen;

            lpbuf->release();
            lpbuf = 0;
            lpmsg->release();
            lpmsg = 0;

            m_pAioImpl->m_pCallbackThread->notify( lpMsg );
            lpMsg = 0;
        }
        else
        {
            lpKey->m_pend_recv_list.push_back( lpMsg );
        }

        if( aio_key::KEY_STATE_RUNNING != lpKey->m_eState )
        {
            lpKey->finish_request( m_pAioImpl );
        }

        return nRet;
    }

    //////////////////////////////////////////////////////////////////////////////////////
    // 

    // 
    // m_bData[0] : true or false for result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : peer port
    // m_szStr[1] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::udp_error( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        cout << lpKey << " error!" << endl;

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // 
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // 
    // m_nData[2] : recv len
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::udp_recv( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        udp_impl * lpUdp = ( udp_impl * )lpKey->m_pHandle;
        socket_impl * lpSocket = &lpUdp->m_socket;
        msg_impl * lpMsg = 0;
        buf_impl * lpBuf = 0;
        bool bIsAddBuf = false;

        if( 0 == nRet )
        {
            lpMsg = lpKey->m_pend_recv_list.front();

            if( !lpMsg )
            {
                g_pXosImpl->create( i_xos::XOS_OBJ_BUF, ( void ** )&lpBuf );
                g_pXosImpl->create( i_xos::XOS_OBJ_MSG, ( void ** )&lpMsg );
                lpKey->m_recv_list.push_back( lpMsg );
                lpMsg->m_lpData[0] = lpKey->m_lpUserKey;
                lpMsg->m_lpData[1] = lpKey;
                lpMsg->m_lpBuf[0] = lpBuf;
                bIsAddBuf = true;
            }
            else
            {
                lpKey->m_pend_recv_list.pop_front();
                lpBuf = lpMsg->m_lpBuf[0];
            }

            lpMsg->m_nData[0] = i_aio::AIO_UDP_RECV_RET;
        }

        if( 0 == nRet )
        {
            char szIp[100] = { 0 };
            xos_u16 usPort = 0;

            lpBuf->m_nLen = lpSocket->udp_recv( lpBuf->m_szBuf, sizeof( lpBuf->m_szBuf ), szIp, &usPort );

            if( lpBuf->m_nLen > 0 )
            {
                strcpy( lpMsg->m_szStr[1], szIp );
                lpMsg->m_nData[1] = usPort;
                lpMsg->m_nData[2] = lpBuf->m_nLen;
                lpMsg->m_bData[0] = true;
            }
            else
            {
                lpMsg->m_bData[0] = false;
            }
        }

        if( 0 == nRet )
        {
            if( !bIsAddBuf )
            {   
                m_pAioImpl->m_pCallbackThread->notify( lpMsg );
            }

            lpMsg = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // 
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // 
    // m_nData[2] : send len
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::udp_send( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        udp_impl * lpUdp = ( udp_impl * )lpKey->m_pHandle;
        socket_impl * lpSocket = &lpUdp->m_socket;
        msg_impl * lpMsg = 0;
        buf_impl * lpBuf = 0;
        int nLastFlag = lpKey->m_nKqueueFlags;
        bool bSave = false;

        if( 0 == nRet )
        {
            if( !lpKey->m_bCanWrite )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            lpMsg = lpKey->m_pend_send_list.front();

            if( lpMsg )
            {
                lpKey->m_pend_send_list.pop_front();
                lpBuf = lpMsg->m_lpBuf[0];
            }
            else
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            if( lpKey->m_nCanWriteSize > lpBuf->m_nLen )
            {
                lpKey->m_nCanWriteSize -= lpBuf->m_nLen;

                if( 0 == lpKey->m_nCanWriteSize )
                {
                    lpKey->m_bCanWrite = false;
                }
            }
            else
            {
                lpKey->m_bCanWrite = false;
                bSave = true;
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            int ret = lpSocket->udp_send( lpBuf->m_szBuf + lpBuf->m_nPos, lpBuf->m_nLen, lpMsg->m_szStr[1], lpMsg->m_nData[1] );

            if( ret == lpBuf->m_nLen )
            {
                lpMsg->m_bData[0] = true;
                lpMsg->m_nData[2] = ret;
            }
            else
            {
                lpKey->m_bCanWrite = false;
                bSave = true;
                nRet = 1;
            }
        }

        if( bSave )
        {
            lpKey->m_pend_send_list.push_front( lpMsg );
            lpMsg = 0;
        }

        if( 0 == nRet )
        {
            m_pAioImpl->m_pCallbackThread->notify( lpMsg );
            lpMsg = 0;
        }

        if( !lpKey->m_bCanWrite || lpKey->m_pend_send_list.front() )
        {
            lpKey->m_nKqueueFlags |= EVFILT_WRITE;

            if( nLastFlag != lpKey->m_nKqueueFlags )
            {
                struct kevent change[1] = { 0 };
                EV_SET( &change[0], lpSocket->m_hSocket, EVFILT_WRITE, EV_ENABLE, 0, 0, lpKey );
                kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
            }
        }
        else
        {
            lpKey->m_nKqueueFlags &= ~EVFILT_WRITE;
            struct kevent change[1] = { 0 };
            EV_SET( &change[0], lpSocket->m_hSocket, EVFILT_WRITE, EV_DISABLE, 0, 0, lpKey );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
        }

        return nRet;
    }

} // xos
